package com.baishancloud.log.fm

import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration

import java.util.{Timer, TimerTask}

/**
 * 过滤延迟超过指定时间的数据
 *
 * @author ziqiang.wang 
 * @date 2022/3/15 17:40
 */
class LateDataFilter(parameterTool: ParameterTool) extends RichFilterFunction[(String, Origin)] {

  var allowLatestSecond: Long = 0L

  override def open(parameters: Configuration): Unit = {
    allowLatestSecond = System.currentTimeMillis() / 1000 - parameterTool.getLong(stateTTL, 1) * 3600
    new Timer("更新最大延迟时间", true).scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = allowLatestSecond = System.currentTimeMillis() / 1000 - parameterTool.getLong(stateTTL, 1) * 3600
    }, 60 * 1000, 60 * 1000)
  }

  override def filter(value: (String, Origin)): Boolean = {
    value._2.datetimeOfUTC().toEpochSecond >= allowLatestSecond
  }
}
